package com.gengzp.mq.client.processor;

import com.gengzp.mq.client.annotations.MqMessageListener;
import com.gengzp.mq.client.core.MqServerRequestFunc;
import com.gengzp.mq.client.interfaces.MqMessageListenerInterface;
import com.gengzp.mq.common.model.resp.GengzpMqMessageExt;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.UUID;

/**
 * @ClassName MqMessageListenerProcessor
 * @Description 消息监听者处理器
 * @Author gengzp
 * @Date 2025/6/9 23:27
 */
@Component
public class MqMessageListenerProcessor implements BeanPostProcessor {

    private static final Logger logger = LoggerFactory.getLogger(MqMessageListenerProcessor.class);

    @Autowired
    private MqServerRequestFunc mqServerRequestFunc;

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        // 只处理带有 @MqMessageListener 注解的类
        MqMessageListener mqMessageListenerAnnotation = bean.getClass().getAnnotation(MqMessageListener.class);
        if (Objects.isNull(mqMessageListenerAnnotation)) {
            return bean;
        }

        // 获取当前监听的topic
        String topic = mqMessageListenerAnnotation.topic();
        if (StringUtils.isBlank(topic)) {
            throw new RuntimeException("类: [" + beanName + "] 使用了 @MqMessageListener 注解, topic 不能为空");
        }

        // 获取当前消费者组, 未填写则给默认
        String consumerGroup = mqMessageListenerAnnotation.consumerGroup();
        if (StringUtils.isBlank(consumerGroup)) {
            consumerGroup = "default";
        }

        // 检查bean是否实现了 MqMessageListenerInterface 接口
        if (!(bean instanceof MqMessageListenerInterface)) {
            throw new RuntimeException("类: [" + beanName + "] 使用了 @MqMessageListener 注解, 必须实现 MqMessageListenerInterface 接口");
        }

        // 执行消费方法
        String finalConsumerGroup = consumerGroup;
        new Thread(() -> {
            while (true) {
                GengzpMqMessageExt gengzpMqMessageExt = null;
                try {
                    gengzpMqMessageExt = mqServerRequestFunc.LongPollingReceiverMessage(topic, finalConsumerGroup);
                } catch (Exception e) {
                    logger.info("监听异常: " + e.getMessage());
                }

                if (Objects.isNull(gengzpMqMessageExt)) {
                    try {
                        Thread.sleep(2000);
                    } catch (Exception ignored) {
                    }
                    continue;
                }

                boolean isSuccess = false;
                try {
                    ((MqMessageListenerInterface) bean).consumeMessage(gengzpMqMessageExt);
                    isSuccess = true;
                } catch (Exception e) {
                    logger.info("消息处理失败: " + e.getMessage());
                } finally {
                    if (isSuccess) {
                        mqServerRequestFunc.confirmMessage(Long.valueOf(gengzpMqMessageExt.getId()));
                    }
                }
            }
        }, String.format("gengzp-mq-listener-%s", UUID.randomUUID())).start();

        return bean;
    }

}
